package drds.global_transaction.resource_manager.undolog;

import drds.common.$;
import drds.common.Constants;
import drds.common.Json;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.common.util.Blobs;
import drds.global_transaction.resource_manager.ConnectionContext;
import drds.global_transaction.resource_manager.model.TableMetaData;
import drds.global_transaction.resource_manager.model.TableMetaDatas;
import drds.global_transaction.resource_manager.proxy.resource_manager.ConnectionProxy;
import drds.global_transaction.resource_manager.proxy.resource_manager.DataSourceProxy;
import drds.global_transaction.resource_manager.undolog.undo_executor.UndoExecutor;
import drds.global_transaction.resource_manager.undolog.undolog_parser.UndoLogParserFactory;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;

import static drds.global_transaction.api.exception.TransactionExceptionCode.branch_transaction_rollback_failed_but_retriable;

@Slf4j
public final class UndoLogManager
{

	private static String UNDO_LOG_TABLE_NAME = "undo_log";

	private static String INSERT_UNDO_LOG_SQL = "INSERT INTO " + UNDO_LOG_TABLE_NAME + "\n" + "\t(branch_id, xid, rollback_info, log_status, log_created, log_modified)\n" + "VALUES (?, ?, ?, 0, now(), now())";
	private static String DELETE_UNDO_LOG_SQL = "DELETE FROM " + UNDO_LOG_TABLE_NAME + "\n" + "\tWHERE branch_id = ? AND xid = ?";

	private static String SELECT_UNDO_LOG_SQL = "SELECT * FROM " + UNDO_LOG_TABLE_NAME + " WHERE log_status = 0 AND " + "branch_id = ? AND xid = ? FOR UPDATE";

	public static void deleteUndoLog(Connection connection, String globalTransactionIdWithAddressAndPort, long branchTransactionId) throws SQLException
	{
		PreparedStatement preparedStatement = connection.prepareStatement(DELETE_UNDO_LOG_SQL);
		preparedStatement.setLong(1, branchTransactionId);
		preparedStatement.setString(2, globalTransactionIdWithAddressAndPort);
		preparedStatement.executeUpdate();
	}

	public static void flushUndoLogs(ConnectionProxy connectionProxy) throws SQLException
	{

		ConnectionContext connectionContext = connectionProxy.getConnectionContext();
		String globalTransactionIdWithAddressAndPort = connectionContext.getGlobalTransactionIdWithAddressAndPort();
		long branchTransactionId = connectionContext.getBranchTransactionId();
		//
		BranchUndoLog branchUndoLog = new BranchUndoLog();
		branchUndoLog.setGlobalTransactionIdWithAddressAndPort(globalTransactionIdWithAddressAndPort);
		branchUndoLog.setBranchTransactionId(branchTransactionId);
		branchUndoLog.setUndoLogList(connectionContext.getUndoLogList());
		String undoLogContent = UndoLogParserFactory.getInstance().encode(branchUndoLog);
		if (Constants.developMode)
		{
			log.debug("undoLogContent: " + undoLogContent);
			log.debug(Json.formate(undoLogContent));
		}
		PreparedStatement preparedStatement = null;
		try
		{
			preparedStatement = connectionProxy.getConnection().prepareStatement(INSERT_UNDO_LOG_SQL);
			preparedStatement.setLong(1, branchTransactionId);
			preparedStatement.setString(2, globalTransactionIdWithAddressAndPort);
			preparedStatement.setBlob(3, Blobs.string2blob(undoLogContent));
			preparedStatement.executeUpdate();
		} catch (Exception e)
		{
			if (e instanceof SQLException)
			{
				throw (SQLException) e;
			} else
			{
				throw new SQLException(e);
			}
		} finally
		{
			if (preparedStatement != null)
			{
				preparedStatement.close();
			}
		}

	}

	public static void undo(DataSourceProxy dataSourceProxy, String globalTransactionIdWithAddressAndPort, long branchTransactionId) throws GlobalTransactionException
	{
		Connection connection = null;
		ResultSet resultSet = null;
		PreparedStatement preparedStatement = null;
		try
		{
			connection = dataSourceProxy.getNativeConnection();
			// The entire undolog process should run in a local transaction.
			connection.setAutoCommit(false);
			// Find UNDO LOG
			preparedStatement = connection.prepareStatement(SELECT_UNDO_LOG_SQL);
			preparedStatement.setLong(1, branchTransactionId);
			preparedStatement.setString(2, globalTransactionIdWithAddressAndPort);
			resultSet = preparedStatement.executeQuery();

			while (resultSet.next())
			{
				Blob blob = resultSet.getBlob("rollback_info");
				String rollbackInfo = Blobs.blob2string(blob);
				BranchUndoLog branchUndoLog = UndoLogParserFactory.getInstance().decode(rollbackInfo);

				for (UndoLog undoLog : branchUndoLog.getUndoLogList())
				{
					TableMetaData tableMetaData = TableMetaDatas.getTableMetaData(dataSourceProxy, undoLog.getTableName());
					undoLog.setTableMeta(tableMetaData);
					UndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(undoLog);
					undoExecutor.executeOn(connection);
				}

			}
			deleteUndoLog(connection, globalTransactionIdWithAddressAndPort, branchTransactionId);// 这里感觉应该是进行标记为待删除可能更好
			connection.commit();

		} catch (Exception e)
		{
			if (Constants.developMode)
			{
				e.printStackTrace();
			}
			log.error($.printStackTraceToString(e));
			if (connection != null)
			{
				try
				{
					connection.rollback();
				} catch (SQLException rollbackEx)
				{
					log.warn("failed to close JDBC resource while undolog ... ", rollbackEx);
				}
			}
			throw new GlobalTransactionException(branch_transaction_rollback_failed_but_retriable, String.format("%s" + "/%s", branchTransactionId, globalTransactionIdWithAddressAndPort), e);

		} finally
		{
			try
			{
				if (resultSet != null)
				{
					resultSet.close();
				}
				if (preparedStatement != null)
				{
					preparedStatement.close();
				}
				if (connection != null)
				{
					connection.close();
				}
			} catch (SQLException closeEx)
			{
				log.warn("failed to close JDBC resource while undolog ... ", closeEx);
			}
		}

	}

	private UndoLogManager()
	{

	}
}
